#include "mstp_types.h"
#include "mstp_proto.h"
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/select.h>
#include <sys/time.h>
#include <sys/types.h>
#include <string.h>
#include <signal.h>
#include <errno.h>
#include <fcntl.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <sys/socket.h>
#include <netinet/tcp.h>
#include <netinet/ip.h>


static int max_atomic = 0;


/*++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++*
 *	FUNCTION PROTOTIPES						    			*			
 *----------------------------------------------------------------------------------------------*
 *      RETURN          FUNCION NAME             ARGUMENTS		       			*
 *++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++*/

PRIVATE chunk_cache_t *	bring_out_chunk		(chunk_cache_t**,u_int32_t);
PRIVATE void 	       	clean_send_buffer	(stream_t *);
/*PRIVATE int 	       	chunk_end         	(stream_t *); */

PRIVATE u_int16_t	choice_chunk_sz		(size_t,int);
PRIVATE ssize_t 	distribute_chunks	(mstp_t *,const void *,size_t);
PRIVATE ssize_t 	flush_all		(mstp_t *,size_t);
PRIVATE int 		stream_rdy		(stream_queue_t *,stream_queue_t *,stream_queue_t*,struct timeval *);
PRIVATE void 		enqueue_rdy_send        (stream_queue_t *,stream_t *);
PRIVATE void 		enqueue_rdy_recv	(stream_queue_t *,stream_t *);
PRIVATE void 		enqueue_	        (stream_queue_t *,stream_t *);
PRIVATE stream_t*	dequeue_		(stream_queue_t *);
PRIVATE stream_t*	dequeue_rdy_send	(stream_queue_t *);
PRIVATE stream_t*	dequeue_rdy_recv	(stream_queue_t *);

PRIVATE int 		stream_connect		(stream_t *,struct sockaddr_in *);
PRIVATE int 		stream_accept		(stream_t *, int);


static size_t 	 put_chunk 		(buff_t *send_buffer, u_int8_t *buffer, u_int16_t sz, u_int32_t seq);
static void     *alloc_chunk		(chunk_cache_t **heap);
static void      free_chunk		(chunk_cache_t *ptr, chunk_cache_t **heap);
static stream_t *new_stream		(u_int32_t recv_buff_len, u_int32_t send_buff_len);
static int       pick_up_chunk_loop	(chunk_cache_t **cache, chunk_cache_t **heap, buff_t *recv_buff);
static int 	 pick_up_chunk		(chunk_cache_t **cache, chunk_cache_t **heap, buff_t *recv_buff);
static ssize_t	 stream_recv		(stream_t *stream);
static int	 get_next_chunk		(mstp_t *socket, chunk_cache_t **chunk);
static ssize_t	 mstp_recv_all_non_block(mstp_t *socket);
static ssize_t 	 mstp_recv_all_block	(mstp_t *socket);
static ssize_t 	 mstp_recv_all		(mstp_t *socket, struct timeval *timeval);
static int 	 full_chunk		(buff_t *recv_buffer);


static void *
alloc_chunk(chunk_cache_t **heap)
{
    chunk_cache_t *ptr;
    if (*heap == NULL) 
    {
	ptr = calloc(1, sizeof(chunk_cache_t));
    }
    else 
    {
	ptr = *heap;
	*heap = ptr->next;
	memset(ptr, 0, sizeof(chunk_cache_t));
    }
    return ptr;
}




static void 
free_chunk(chunk_cache_t *ptr, chunk_cache_t **heap)
{
    if (ptr != NULL)
    {
	ptr->prev = NULL;
	ptr->next = *heap;
	*heap = ptr;
    }
}


PUBLIC int mstp_accept (mstp_t *mstp_socket, int streams)
{
    mstp_proto_t 	proto;
    struct sockaddr_in	rhost_addr;
    ssize_t  		recv_bytes;
    u_int8_t 		nr_streams;
    stream_t 		*stream;
    u_int32_t 		r_buff_len;
    u_int32_t 		s_buff_len;
    socklen_t		len_addr;
    int 		socket_tcp;
    int 		yes;
    
    if (streams > mstp_socket->max_streams)
	return -1;
	

    len_addr = sizeof(struct sockaddr_in);	
	
    recv_bytes = recvfrom(mstp_socket->udp_socket, 
		       (void *)&proto, 
		       sizeof(mstp_proto_t),
		       0, 
		       (struct sockaddr *)&rhost_addr, 
		       &len_addr);
		       
		       
    if (recv_bytes == -1 || recv_bytes < sizeof(mstp_proto_t))
	return -1;
    	
    if (proto.msg_type == CNT) {
	nr_streams = (proto.msg_body.cnt.streams > streams) ? streams : proto.msg_body.cnt.streams;
	proto.msg_type = ACK;
	proto.msg_body.ack.streams = nr_streams;
	proto.msg_body.ack.addr	   = mstp_socket->addr.sin_addr.s_addr;
	proto.msg_body.ack.port    = mstp_socket->addr.sin_port;
	

	
	socket_tcp = socket(PF_INET, SOCK_STREAM, 0);
	if (socket_tcp == -1)
	    return -1;

	yes = 1;	
	if (setsockopt(socket_tcp, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(int)) == -1)
	    return -1;
	    
	if (bind(socket_tcp, (struct sockaddr *)&(mstp_socket->addr), sizeof(struct sockaddr_in)) == -1)
	    return -1;
	
	if (listen(socket_tcp, 5) == -1)
	    return -1;
	    	
	if (sendto(mstp_socket->udp_socket, &proto, sizeof(mstp_proto_t),0,(struct sockaddr *)&rhost_addr,sizeof(struct sockaddr_in) ) == -1)
	    return -1;    	
	    	
	r_buff_len = MAX_BUFF_LEN / nr_streams;
	s_buff_len = MAX_BUFF_LEN / nr_streams;
	
	mstp_socket->nr_streams = nr_streams;
	
	while ( nr_streams > 0 ) {
	
	    stream = new_stream(r_buff_len, s_buff_len);
	    if (stream != NULL) {
		if (stream_accept(stream, socket_tcp) == -1) 
		    return -1;
		
		enqueue_(&(mstp_socket->streams), stream);
		nr_streams--;
	    }	
	    else {
		/*
		 * Fuga de memoria
		 */
		return -1;
	    }
	
	}
	close(socket_tcp);

    }
    else {
	return -1;
    }
    return 0;
}


PUBLIC int mstp_bind(mstp_t *socket, struct sockaddr_in *addr)
{
    memcpy(&(socket->addr), addr, sizeof(struct sockaddr_in));
    return bind(socket->udp_socket, (struct sockaddr *) addr, sizeof(struct sockaddr_in));
}

PUBLIC int mstp_connect(mstp_t *mstp_socket, struct sockaddr_in *addr, struct timeval *timeout)
{
    struct sockaddr_in	host_addr;
    mstp_proto_t 	proto;
    stream_t	 	*stream;
    fd_set 		recvset;
    socklen_t		len_addr;
    u_int32_t		r_buff_len;
    u_int32_t		s_buff_len;
    u_int8_t  		nr_s;
    int 		ret_val;
    
    
    FD_ZERO(&recvset);
        
    proto.msg_type    	       = CNT;
    proto.msg_body.cnt.streams = 50;
    
    ret_val = sendto(mstp_socket->udp_socket, &proto, sizeof(mstp_proto_t), 0, (struct sockaddr *) addr, sizeof(struct sockaddr_in));
    if ( ret_val == -1 ) 
	return -1;
	
    
    FD_SET(mstp_socket->udp_socket, &recvset);
    ret_val = select(mstp_socket->udp_socket+1,&recvset, NULL, NULL, timeout); 
    if ( ret_val == 0 ) {
	/*
	 * Timeout
	 */
    
	return -1;
    }
    if (FD_ISSET(mstp_socket->udp_socket, &recvset)) {
	len_addr = sizeof(struct sockaddr_in);
	ret_val = recvfrom(mstp_socket->udp_socket, &proto, sizeof(mstp_proto_t), 0,(struct sockaddr *) &host_addr,&len_addr);
    
	if ( ret_val == -1 )
	    return -1;

	
	if ((addr->sin_port == host_addr.sin_port) && (addr->sin_addr.s_addr == host_addr.sin_addr.s_addr )) {
	    /*
	     * Al mismo que le envie me esta respondiendo
	     */
	    if (proto.msg_type == ACK) {
	
		nr_s = proto.msg_body.ack.streams;

		mstp_socket->nr_streams = nr_s;

		host_addr.sin_port 		= proto.msg_body.ack.port;
		host_addr.sin_addr.s_addr 	= proto.msg_body.ack.addr;		
	    
		r_buff_len = MAX_BUFF_LEN / nr_s;
		s_buff_len = MAX_BUFF_LEN / nr_s;
	    
		while (nr_s > 0) {
		
		    stream = new_stream(r_buff_len, s_buff_len);
		    if (stream == NULL)
			return -1;
				
		    if (stream_connect(stream,&host_addr)== -1)
			return -1;
		    
		    
		    enqueue_(&(mstp_socket->streams), stream);    
	
		    nr_s--;	
		}
	    }
	    else {
		return -1;
	    }
	}
	else {
	    return -1;
	}
    }
    else {
	return -1;
    }
    return 0;
}


/*++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++*
 *											*
 * new_stream(recv_buff_len, send_buff_len):  Crea la estructura de datos stream	*
 *											*
 * Retorno: - Puntero a la nueva estructura stream_t recientemente creada		*
 *          - NULL en error								*
 *++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++*/
static stream_t *
new_stream(u_int32_t recv_buff_len, u_int32_t send_buff_len)
{
    stream_t *stream;
    stream = (stream_t *)calloc(1, sizeof(stream_t));
    
    if (stream != NULL) 
    {
	stream->socket         = 0;
	stream->bytes_send     = 0;
	stream->bytes_recv     = 0;
	stream->nr_rdy_s_times = 0;
	stream->nr_rdy_r_times = 0;
	stream->bytes_in_queue = 0;
    	if (gettimeofday(&(stream->start_time), NULL)==-1)
    	    return NULL;
	stream->average_speed = 0;
	stream->p_nextready_send = NULL;
	stream->p_nextready_recv = NULL;
	stream->p_next		 = NULL;
    
	stream->send_buffer.buff = (u_int8_t *) calloc(send_buff_len, 1); 
        if (stream->send_buffer.buff == NULL)
	    return NULL;    
        stream->recv_buffer.buff = (u_int8_t *) calloc(recv_buff_len, 1);
	if (stream->recv_buffer.buff == NULL) 
	{
	    free(stream->send_buffer.buff);
	    return NULL;
	}
	
	stream->send_buffer.buff_sz = stream->send_buffer.bytes_free = send_buff_len;
	stream->recv_buffer.buff_sz = stream->recv_buffer.bytes_free = recv_buff_len;
	stream->send_buffer.bytes_occ = 0;
	stream->recv_buffer.bytes_occ = 0;	
	stream->send_buffer.buff_free = &(stream->send_buffer.buff)[0];
	stream->recv_buffer.buff_free = &(stream->recv_buffer.buff)[0];
	
	return stream;
    }
    return NULL;
}


PRIVATE int stream_accept(stream_t *stream, int socket_tcp)
{
    u_int32_t buff_size;
    int flag, tos;
    
    stream->socket = accept(socket_tcp, (struct sockaddr *) 0,( socklen_t *) 0);
    if (stream->socket == -1)
	return -1;

    flag = 0;
    
    if (setsockopt(stream->socket, IPPROTO_TCP, TCP_CORK, &flag, sizeof(int) ) == -1)
	return -1;
	
    flag = 1;

    tos = IPTOS_THROUGHPUT;

    setsockopt(stream->socket, IPPROTO_IP, IP_TOS, &tos, sizeof(tos));
    
    if (setsockopt(stream->socket, IPPROTO_TCP, TCP_NODELAY, &flag, sizeof(int) ) == -1)
	return -1;

    if (fcntl(stream->socket, F_SETFL, O_NONBLOCK) == -1)
	return -1;

    buff_size = 1024*1024*4;

    if (setsockopt(stream->socket, SOL_SOCKET, SO_RCVBUF, &buff_size, sizeof(u_int32_t)) == -1)
	return -1;
	 
    if (setsockopt(stream->socket, SOL_SOCKET, SO_SNDBUF, &buff_size, sizeof(u_int32_t)) == -1)
	return -1;

    stream->status = STREAM_OPEN;

    return 0;
}


PRIVATE int stream_connect(stream_t *stream,struct sockaddr_in *addr)
{
    u_int32_t buff_size;
    int flag;
    int tos;
    
    stream->socket = socket(PF_INET, SOCK_STREAM, 0);
    if (stream->socket == -1)
	return -1;

    buff_size = 1024*4096;
    
    if (connect(stream->socket,(struct sockaddr *)addr, sizeof(struct sockaddr_in))==-1)
	return -1;
    
    flag = 1;
    
    if (setsockopt(stream->socket, IPPROTO_TCP, TCP_CORK, &flag, sizeof(int) ) == -1)
	return -1;
	
    flag = 1;

    tos = IPTOS_THROUGHPUT;

    setsockopt(stream->socket, IPPROTO_IP, IP_TOS, &tos, sizeof(tos));
    
    if (setsockopt(stream->socket, IPPROTO_TCP, TCP_NODELAY, &flag, sizeof(int) ) == -1)
	return -1;
        
    if (fcntl(stream->socket, F_SETFL, O_NONBLOCK) == -1)
	return -1;

    if (setsockopt(stream->socket, SOL_SOCKET, SO_RCVBUF, &buff_size, sizeof(u_int32_t)) == -1)
	return -1;
	 
    if (setsockopt(stream->socket, SOL_SOCKET, SO_SNDBUF, &buff_size, sizeof(u_int32_t)) == -1)
	return -1;

    stream->status = STREAM_OPEN;

    return 0;
}    

static int
iptos_throughput(int socket)
{
    int tos = IPTOS_THROUGHPUT;
    return setsockopt(socket, IPPROTO_IP, IP_TOS, &tos, sizeof(int));
}

static int
tcp_cork(int socket)
{
    int flag = 1;
    return setsockopt(socket, IPPROTO_TCP, TCP_CORK, &flag, sizeof(int));
}

static int
tcp_nodelay(int socket)
{
    int flag = 1;
    return setsockopt(socket, IPPROTO_TCP, TCP_NODELAY, &flag, sizeof(int));
}

static int
tcp_rcvbuf(int socket, u_int32_t buff_size )
{
    return setsockopt(socket, SOL_SOCKET, SO_RCVBUF, &buff_size, sizeof(u_int32_t));
}

static int
tcp_sndbuf(int socket, u_int32_t buff_size )
{
    return setsockopt(socket, SOL_SOCKET, SO_SNDBUF, &buff_size, sizeof(u_int32_t));
}


static void 
enqueue_(stream_queue_t *queue, stream_t *ptr)
{

    ptr->p_next = NULL;
    if (queue->tail == NULL && queue->head == NULL)
        queue->head = ptr;
    else
        queue->tail->p_next = ptr;	
    queue->tail = ptr;
}

static stream_t *
dequeue_(stream_queue_t *queue)
{
    struct stream *ptr;
    
    ptr = queue->head;
    if (ptr != NULL) {
	queue->head = ptr->p_next;
	if (queue->head == NULL)
	    queue->tail = NULL;

    }
    return ptr;
}

static stream_t *
dequeue_rdy_send(stream_queue_t *queue)
{
    struct stream *ptr;
    
    ptr = queue->head;
    if (ptr) {
	queue->head = ptr->p_nextready_send;
	ptr->p_nextready_send = NULL;
	if (queue->head == NULL)
	    queue->tail = NULL;
    }
    return ptr;
}

static stream_t *
dequeue_rdy_recv(stream_queue_t *queue)
{
    struct stream *ptr;
    
    ptr = queue->head;
    if (ptr) 
    {
	queue->head = ptr->p_nextready_recv;
	ptr->p_nextready_recv = NULL;
	if (queue->head == NULL)
	    queue->tail = NULL;
    }
    return ptr;
}


static void 
enqueue_rdy_send(stream_queue_t *queue, stream_t *ptr)
{
    ptr->p_nextready_send = NULL;
    if (queue->tail == NULL && queue->head == NULL)
        queue->head = ptr;
    else
        queue->tail->p_nextready_send = ptr;	
    queue->tail = ptr;
}


static void 
enqueue_rdy_recv(stream_queue_t *queue, stream_t *ptr)
{
    ptr->p_nextready_recv = NULL;
    if (queue->tail == NULL && queue->head == NULL)
        queue->head = ptr;
    else
        queue->tail->p_nextready_recv = ptr;	
    queue->tail = ptr;
}


static ssize_t 
stream_recv(stream_t *stream)
{
    ssize_t    ret = 0;
    u_int8_t  *msg_ptr;
    u_int32_t  bytes_free;
    
    bytes_free = stream->recv_buffer.bytes_free;
    msg_ptr    = stream->recv_buffer.buff_free; 

    ret = recv(stream->socket, msg_ptr, bytes_free, MSG_DONTWAIT);	
    if (ret > 0) 
    {
	if (ret > max_atomic)
	    max_atomic = ret;
	    
	stream->recv_buffer.bytes_free -= ret;
	stream->recv_buffer.bytes_occ  += ret;
	stream->recv_buffer.buff_free   = &(stream->recv_buffer.buff_free)[ret];
	stream->bytes_recv += ret;
    }
    return ret;
}

print_max_atomic(void)
{
    printf ("Max Atomic: %d\n", max_atomic);
}


static ssize_t 
mstp_recv_all_non_block(mstp_t *socket)
{
    struct timeval tv;
    tv.tv_sec  = 0;
    tv.tv_usec = 0;
    
    return mstp_recv_all(socket, &tv);
}

static ssize_t 
mstp_recv_all_block (mstp_t *socket)
{
    return mstp_recv_all(socket,NULL);
}



static ssize_t 
mstp_recv_all (mstp_t *socket, struct timeval *timeout)
{
    stream_t 	   *ptr;
    stream_queue_t recvq;
    u_int32_t	   total_recv, last_recv;

    int eos, n;

    recvq.head = NULL;
    recvq.tail = NULL;
    
    if ((n=stream_rdy(&(socket->streams), &recvq, NULL,timeout))==-1)
	return -1;
    else 
    {
	if (n==0)
	    return 0;
    }
    
    total_recv = 0;
    
    ptr = dequeue_rdy_recv(&recvq);
    while ( ptr != NULL ) 
    {
        eos = 0;	
	last_recv = stream_recv(ptr);
	
	if ( last_recv > 0) 
	    total_recv += last_recv;
	else 
	{
	    if ( last_recv == 0 )
		eos = 1;
	    else
		return -1;
	}
	
	if (pick_up_chunk_loop(&(socket->recv_cache), &(socket->heap), &(ptr->recv_buffer)) == -1)
	    return -1;	
	
	if (eos)
	{
	    if (ptr->recv_buffer.bytes_occ != 0) 
		return -1;
	    else
		ptr->status = STREAM_CLOSE;
	}
	ptr = dequeue_rdy_recv(&recvq);
    }
    return total_recv;
}		


static int 
status_open(stream_queue_t *queue)
{
    int counter = 0;
    stream_t *ptr;
    
    ptr = queue->head;
    while (ptr != NULL)
    {
	if (ptr->status == STREAM_OPEN)
	    counter++;
	ptr = ptr->p_next;
    }
    return counter;
}

static int 
pick_up_chunk_loop(chunk_cache_t **cache, chunk_cache_t **heap, buff_t *recv_buff)
{
    while (full_chunk(recv_buff))
	if (pick_up_chunk(cache, heap, recv_buff) == -1)
	    return -1;

    return 0;
}

static int 
get_next_chunk(mstp_t *socket, chunk_cache_t **chunk)
{
    ssize_t     last_recv;
    u_int32_t	chunk_nr;
    
    chunk_nr = (socket->chunk_seq_recv)++;
    *chunk   = bring_out_chunk(&(socket->recv_cache),chunk_nr);
    
    if (*chunk == NULL) 
    {
	while (*chunk == NULL)
	{
	    last_recv = mstp_recv_all_block(socket);
	    if (last_recv == 0) 
	    {
	        if (status_open(&(socket->streams))==0)
	            break;
	    }	
	    else
	    {
		if (last_recv == -1) {
		    return -1;
		}    
	    }
	    *chunk = bring_out_chunk(&(socket->recv_cache),chunk_nr);
	}
    }
    else 
    {
	if (mstp_recv_all_non_block(socket) == -1)
	    return -1;		
    }

    return 0;
}


static chunk_cache_t *
bring_out_chunk (chunk_cache_t **cache, u_int32_t chunk_nr)
{
    chunk_cache_t *ptr;
    chunk_cache_t *ret = NULL;

    ptr = *cache;
   
    if (ptr != NULL) 
    {
	    
	while (ptr->hdr.chunk_nr < chunk_nr && ptr->next != NULL) 
	    ptr = ptr->next;

	if (ptr->hdr.chunk_nr == chunk_nr) 
	{
	    /*
	     * Lo encontro
	     */
	    ret = ptr;
	    if (*cache == ptr)
		*cache = ptr->next;
	    if (ptr->next != NULL) 
		ptr->next->prev = ptr->prev;
	    if (ptr->prev != NULL)
		ptr->prev->next = ptr->next;

	    ret->prev = NULL;
	    ret->next = NULL;
	}
    }
    return ret;
}


static chunk_cache_t *
get_first_chunk (chunk_cache_t **cache)
{
    chunk_cache_t *ptr;
    
    ptr = *cache;
    if (ptr) 
    {
	*cache = ptr->next;
	if (ptr->next != NULL)
	    ptr->next->prev = ptr->prev;
	
	ptr->prev = NULL;
	ptr->next = NULL;
    
    }
    return ptr;
}


void 
clean_send_buffer (stream_t * stream)
{
    if ( stream != NULL ) 
    {
	if (stream->bytes_in_queue == 0) 
	{
	    stream->send_buffer.bytes_occ  = 0;
	    stream->send_buffer.bytes_free = stream->send_buffer.buff_sz;
	    stream->send_buffer.buff_free  = &(stream->send_buffer.buff)[0];
	    memset(stream->send_buffer.buff, 0, stream->send_buffer.buff_sz);
	    return;
	}
    }
    return;
}

int 
full_chunk(buff_t *recv_buffer)
{
    chunk_hdr_t *hdr;
    
    if (recv_buffer != NULL)
    {
	if (recv_buffer->bytes_occ > CHUNK_HDR_SZ) 
	{
	    hdr = (chunk_hdr_t *) &(recv_buffer->buff)[0];
	    if (hdr->chunk_sz <= (recv_buffer->bytes_occ - CHUNK_HDR_SZ)) {
		return 1;
	    }
	}
    
    }
    return 0;    
}

/*
PRIVATE int chunk_end(stream)
stream_t *stream;
{
    chunk_hdr_t *hdr;
    
    if (stream != NULL) {
	if (stream->recv_buffer.bytes_occ > CHUNK_HDR_SZ) {
	    hdr = (chunk_hdr_t *) &(stream->recv_buffer.buff)[0];
	    if (hdr->chunk_sz <= (stream->recv_buffer.bytes_occ - CHUNK_HDR_SZ))
		return 1;
	}
    }	
    return 0;
}
*/

static void 
insert_chunk_in_cache (chunk_cache_t **cache, chunk_cache_t *new_node)
{
    chunk_cache_t *ptr;

    ptr = *cache;

    if (ptr == NULL) 
    {
	new_node->next = NULL;
	new_node->prev = NULL;
	*cache = new_node;
    }
    else 
    {
	/*
         * Busca la posicion
         */    
	while ( ptr->hdr.chunk_nr < new_node->hdr.chunk_nr  && ptr->next != NULL)
	    ptr = ptr->next;

	if (ptr->hdr.chunk_nr > new_node->hdr.chunk_nr) 
	{
	    if (ptr->prev != NULL)
		ptr->prev->next = new_node;
	    
	    new_node->next  = ptr;
	    new_node->prev  = ptr->prev;
	    ptr->prev	= new_node;
	    if (ptr == *cache)
		*cache = new_node;
	
	}
        else {
	    if (ptr->next == NULL) 
	    {
		new_node->next = NULL;
	        new_node->prev = ptr;
	        ptr->next = new_node;
	    }
	}
    }
}

/*+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++*
 *                                                                                     *
 * pick_up_chunk(cache, stream): Lee de buffer un chunk y lo guarda en el cache        *
 * 										       *
 * Retorno: -  0 En exito							       * 
 *          - -1 En caso de que falla						       *
 *+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++*/  
static int 
pick_up_chunk (chunk_cache_t **cache, chunk_cache_t **heap, buff_t *recv_buff)
{
    chunk_hdr_t   *hdr;
    chunk_cache_t *ptr;
    
    ptr = alloc_chunk(heap);
    
    if (ptr != NULL) 
    {
	/*
	 * Crea el chunk
	 */
	memcpy(&(ptr->hdr),&(recv_buff->buff)[0], CHUNK_HDR_SZ);
	hdr = &(ptr->hdr);
	memcpy(ptr->data,&(recv_buff->buff)[CHUNK_HDR_SZ], hdr->chunk_sz);

	/*
	 * Lo inserta en la lista
	 */
	insert_chunk_in_cache(cache, ptr);
	
	recv_buff->bytes_occ  -= (CHUNK_HDR_SZ + hdr->chunk_sz);
	memmove(recv_buff->buff, &(recv_buff->buff)[CHUNK_HDR_SZ+hdr->chunk_sz], recv_buff->bytes_occ);
	recv_buff->bytes_free += (CHUNK_HDR_SZ + hdr->chunk_sz);
	recv_buff->buff_free   = (u_int8_t *) (&(recv_buff->buff_free)[0] - (u_int8_t*) (CHUNK_HDR_SZ + hdr->chunk_sz));
    
	return 0;
    }
    return -1;
}


int 
mstp_socket (mstp_t *mstp_socket)
{
    mstp_socket->io_buffer.buff = (u_int8_t *) calloc(1, MAX_BUFF_LEN);
    if ( mstp_socket->io_buffer.buff ) 
    {
    
	mstp_socket->udp_socket = socket(PF_INET, SOCK_DGRAM, 0);
    
	if ( mstp_socket->udp_socket != -1 ) 
	{
    
	    mstp_socket->io_buffer.bytes_occ  = 0;
	    mstp_socket->io_buffer.bytes_free = MAX_BUFF_LEN;
	    mstp_socket->io_buffer.buff_sz    = MAX_BUFF_LEN;
	    mstp_socket->io_buffer.buff_free  = &(mstp_socket->io_buffer.buff)[0];	
	
	    mstp_socket->chunk_seq_send = 0;
	    mstp_socket->chunk_seq_recv = 0;
	    mstp_socket->nr_streams     = 0;

	    mstp_socket->heap  	      = NULL;
	    mstp_socket->recv_cache   = NULL;
	    mstp_socket->send_cache   = NULL;
	    mstp_socket->streams.head = NULL;
    	    mstp_socket->streams.tail = NULL;
	    mstp_socket->max_streams  = DEFAULT_MAX_STREAMS;

	    memset(&(mstp_socket->addr), 0, sizeof(struct sockaddr_in));
	
	    return  0;
	}
	else 
	{
	    return -1;
	}
    }
    else 
    {
    
	return -1;
    }
}


static u_int16_t 
choice_chunk_sz (size_t len, int nr_streams)
{
    u_int32_t cz;
    int div = 0;
    do 
    {
	div = div + nr_streams; 
	cz  = len / div;
    } while ( cz > MAX_CHUNK_SZ);

    return cz;
}

 
static size_t
put_chunk (buff_t *send_buffer, u_int8_t *buffer, u_int16_t sz, u_int32_t seq)
{
    chunk_hdr_t chdr;

    chdr.chunk_sz = sz;
    chdr.chunk_nr = seq;

    memcpy(send_buffer->buff_free, &chdr, CHUNK_HDR_SZ);
    memcpy(&(send_buffer->buff_free)[CHUNK_HDR_SZ], buffer, sz);
    
    send_buffer->bytes_free -= (sz + CHUNK_HDR_SZ);
    send_buffer->bytes_occ  += (sz + CHUNK_HDR_SZ);
    send_buffer->buff_free   = &((send_buffer->buff_free)[sz + CHUNK_HDR_SZ]);

    return (sz + CHUNK_HDR_SZ);
}


PUBLIC ssize_t mstp_send(mstp_t *socket, const void *buffer, size_t len)
{
    ssize_t total_len;
    
    if ( (total_len = distribute_chunks(socket,buffer,len)) == -1 )
	return (ssize_t) -1;
	
    return flush_all(socket, total_len);    
}


PUBLIC ssize_t mstp_recv(mstp_t *socket, void *buffer, size_t len)
{
    chunk_cache_t *chunk;
    
    
    if ( len > ATOMIC_RECV )
	len = ATOMIC_RECV;
    
    chunk = NULL;
    
    while (socket->io_buffer.bytes_occ < len) {

	if (get_next_chunk(socket, &chunk) == -1)
	    return (ssize_t) -1;
	
	if (chunk == NULL) {
	    if (socket->recv_cache != NULL) {
		return (ssize_t) -1;
	    }
	    else
		break;
	}
	else {
	    memcpy(socket->io_buffer.buff_free, chunk->data, chunk->hdr.chunk_sz);
	    socket->io_buffer.bytes_free -= (chunk->hdr.chunk_sz);
	    socket->io_buffer.bytes_occ  += (chunk->hdr.chunk_sz);
	    socket->io_buffer.buff_free   = &(socket->io_buffer.buff_free)[chunk->hdr.chunk_sz];
	    free_chunk(chunk, &(socket->heap));
	    chunk = NULL;
	}
    }

    if (socket->io_buffer.bytes_occ < len) {
	
	/*
	 * No hay mas nada que recibir --- Se cerraron los sockets
	 */
	len = socket->io_buffer.bytes_occ;
    }
    
    if (len > 0) {
	memcpy(buffer, socket->io_buffer.buff, len);
        memmove(socket->io_buffer.buff, &(socket->io_buffer.buff)[len], socket->io_buffer.bytes_occ - len);
        socket->io_buffer.bytes_occ  = socket->io_buffer.bytes_occ - len;
        socket->io_buffer.bytes_free = socket->io_buffer.buff_sz   - socket->io_buffer.bytes_occ;
        socket->io_buffer.buff_free  = &(socket->io_buffer.buff)[socket->io_buffer.bytes_occ];
    }
    return len;
}


static ssize_t
distribute_chunks (mstp_t *socket, const void *buffer, size_t len)
{
    size_t        new_len, mod;
    ssize_t       total_len;
    u_int16_t     cz;
    u_int32_t 	  num_chunks;
    chunk_cache_t *new_chunk;
    u_int8_t      *buff_ptr = &((u_int8_t *)buffer)[0];
    
    new_len = ( len > ATOMIC_SEND ) ? (size_t) ATOMIC_SEND : len;
    
    cz = choice_chunk_sz(new_len,socket->nr_streams);
    num_chunks = new_len / cz;
    mod        = new_len % cz;
    if ( mod )
	num_chunks++;
    
    
    total_len = (num_chunks * CHUNK_HDR_SZ) + len;
    
    while (num_chunks > 0)    
    {
	new_chunk = alloc_chunk(&(socket->heap));
	if (new_chunk != NULL)
	{
	    if (num_chunks == 1 && mod)
	    {
		new_chunk->hdr.chunk_sz = mod;	
		memcpy(new_chunk->data, buff_ptr, mod);
		buff_ptr = &buff_ptr[mod];
	    }
	    else
	    {
		new_chunk->hdr.chunk_sz = cz;
		memcpy(new_chunk->data, buff_ptr, cz);
		buff_ptr = &buff_ptr[cz];
	    }
	    new_chunk->hdr.chunk_nr = (socket->chunk_seq_send)++;
	    insert_chunk_in_cache(&(socket->send_cache), new_chunk);
	}
	else
	{
	    return (ssize_t) -1;
	}
	num_chunks--;
    }
    return total_len;
}

static ssize_t
flush_all(mstp_t *socket, size_t len)
{
    stream_queue_t sendq;
    stream_t       *ptr;
    ssize_t        last_send;
    size_t	   bytes_sent;
    
    u_int32_t total, wsend, total_chunks;
    u_int8_t  	  *msg_ptr;
    chunk_cache_t *chunk;
    
    sendq.head = NULL;
    sendq.tail = NULL;
    
    bytes_sent   = 0;
    total_chunks = 0;
    
    while (bytes_sent < len)
    {
	if (stream_rdy(&(socket->streams), NULL, &sendq, NULL) == -1)
	    return (ssize_t) -1;
	
	ptr = dequeue_rdy_send(&sendq);
	    
	while (ptr != NULL && bytes_sent < len ) 
	{
	    if ( ptr->bytes_in_queue == 0 )
	    {
		clean_send_buffer(ptr);
		chunk = get_first_chunk(&(socket->send_cache));
		if ( chunk != NULL )
		{
		    ptr->bytes_in_queue += put_chunk(&(ptr->send_buffer), chunk->data, chunk->hdr.chunk_sz, chunk->hdr.chunk_nr); 
		    total_chunks++;
		    free_chunk(chunk, &(socket->heap));
		}
	    }
	
	    total = ptr->send_buffer.bytes_occ;
	    wsend = ptr->bytes_in_queue;
	    msg_ptr = &(ptr->send_buffer.buff)[total-wsend];
	    if (wsend > 0)
	    {
		last_send = send(ptr->socket, msg_ptr, wsend, MSG_DONTWAIT | MSG_NOSIGNAL);
		if (last_send > 0) 
		{
		    bytes_sent          = bytes_sent          + last_send;
		    ptr->bytes_in_queue = ptr->bytes_in_queue - last_send;
		    ptr->bytes_send     = ptr->bytes_send     + last_send;
		}			     
		else 
		{
		    if ( last_send == -1 ) 
		    {
			return -1;
		    } 
		}    
	    }
	    ptr = dequeue_rdy_send(&sendq);
	}
    }
    return bytes_sent - (total_chunks * CHUNK_HDR_SZ);
}    

    
static int 
stream_rdy(stream_queue_t *stream_queue, stream_queue_t *recv, stream_queue_t *send, struct timeval *timeout)
{
    int retval = 0;
    int maxfd  = -1;
    int counter = 0;
    int n;
    stream_t *ptr = NULL;
    fd_set recvset;
    fd_set sendset;
    fd_set *recvts;
    fd_set *sendts;
    FD_ZERO(&recvset);
    FD_ZERO(&sendset);
    
    if (stream_queue != NULL) 
    {
	ptr = stream_queue->head;
	while ( ptr != NULL ) 
	{
	    maxfd = (ptr->socket > maxfd) ? ptr->socket : maxfd;
	    if (ptr->status == STREAM_OPEN) 
	    {
	        FD_SET(ptr->socket, &recvset);
		FD_SET(ptr->socket, &sendset);
		counter++;
	    }
	    ptr = ptr->p_next;
	}
	recvts = (recv != NULL) ? &recvset : NULL;
    	sendts = (send != NULL) ? &sendset : NULL;
	/*
	 * ToDo: This section needs to be ATOMIC (sigprocmask)
	 */ 
	
	
	if (counter == 0) 
	    return 0;
	 
	n = select(maxfd+1, recvts, sendts, NULL, timeout);
	if (n == -1)
	    retval = -1;
	else 
	{
	    retval = n;
	    if ( n > 0 && recv != NULL ) 
	    {
		ptr = stream_queue->head;
		recv->head = NULL;
		recv->tail = NULL;
		while ( ptr != NULL ) 
		{
		    if (FD_ISSET(ptr->socket, recvts)) 
		    {
			n--;
			retval++;
			ptr->nr_rdy_r_times++;
			enqueue_rdy_recv(recv, ptr);
		    }
		    ptr = ptr->p_next;
		    if (n == 0)
			break;
		}
	    }
	    if ( n > 0 && send != NULL ) 
	    {
		ptr = stream_queue->head;
		send->head = NULL;
		send->tail = NULL;
		while ( ptr != NULL ) 
		{
		    if (FD_ISSET(ptr->socket, sendts)) 
		    {
			n--;
			retval++;
			ptr->nr_rdy_s_times++;
			enqueue_rdy_send(send, ptr);
		    }
		    ptr = ptr->p_next;
		    if ( n == 0 )
			break;
		}  
	    }
	} /* else */
    } /* if (stream != NULL) */
    return retval;
}    
